package saic.spark_stream

import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.kafka.KafkaUtils
import kafka.serializer.StringDecoder

/**
 * @author ZhiLi
 */
object StreamHello {
  def main(args: Array[String]): Unit = {


    // Create context with 2 second batch interval
    val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local")
    val ssc = new StreamingContext(sparkConf,Seconds(1))
    
    //copy code
    val kafkaParams = Map("metadata.broker.list"->"192.168.40.129:9092")
    val topics = Set("mykafka")

    val kafkaStream = KafkaUtils.createDirectStream[String, String, 
      StringDecoder,StringDecoder](ssc, kafkaParams, topics)
    val flatMapStream = kafkaStream.flatMap(_._2)

    flatMapStream.print()
    //flatMapStream.persist()
    
    
    
//################################################################################
    //My code
    // Create direct kafka stream with brokers and topics
//    val topicsSet = "mykafka".split(",").toSet
//    val brokers = "192.168.40.129:9092"
//    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
//    val messages = KafkaUtils.createDirectStream(ssc, kafkaParams, topicsSet)
    // Get the lines, split them into words, count the words and print
//    val lines = messages.map(_.toString())
//    lines.print()
//    val words = lines.flatMap(_.split(" "))
//    val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
//    wordCounts.print()
//##################################################################################

    // Start the computation
    ssc.start()
    ssc.awaitTermination()
  }
}